package com.river.userdraw;

import com.google.common.base.Charsets;
import com.google.common.base.Splitter;
import com.google.common.collect.Maps;
import com.google.common.io.Files;
import org.apache.commons.lang.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.io.File;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.stream.Collectors;


/**
 * @author riverfan
 */
public class UserDrawDemon {
    private static final Splitter SPLITTER = Splitter.on('|');
    public static void main(String[] args) throws IOException {
        UserDrawConfig userDrawConfig = UserDrawConfig.getInstance();
        System.out.println(userDrawConfig);
        SparkConf conf = new SparkConf();
        conf.setAppName("UserDrawDemon");
        //设置master属性
        conf.setMaster("local[2]");
        JavaSparkContext sc = new JavaSparkContext(conf);
        String userdataFile = "/Users/riverfan/mytest/spark/userDraw/userdata.txt";


        File file = new File("/Users/riverfan/mytest/spark/userDraw/appTab.txt");

        //读取文件  内容比较少 所以直接读取了
        Map<String, AppUserWeightVo> appMap = Files.readLines(file, Charsets.UTF_8)
                .stream()
                .map(t -> {
                    List<String> list = SPLITTER.splitToList(t);
                    return new AppUserWeightVo(
                            list.get(0),
                            list.get(1),
                            Float.parseFloat(list.get(2)),
                            Float.parseFloat(list.get(3)),
                            Float.parseFloat(list.get(4)),
                            Float.parseFloat(list.get(5)),
                            Float.parseFloat(list.get(6)),
                            Float.parseFloat(list.get(7)),
                            Float.parseFloat(list.get(8)));
                }).collect(Collectors.toMap(AppUserWeightVo::getAppId, t -> t));

        Map<String, UserDrawVo> userDrawMap = Maps.newHashMap();
        sc.textFile(userdataFile)
                .map(t -> {
                    List<String> list = SPLITTER.splitToList(t);
                    String now = list.get(userDrawConfig.getDateIndex());
                    UserInfoVo userInfoVo = new UserInfoVo();
                    userInfoVo.setNow(new SimpleDateFormat("yyyyMMdd").format(Long.parseLong(now)));
                    userInfoVo.setMobile(list.get(userDrawConfig.getMdnIndex()));
                    userInfoVo.setAppId(list.get(userDrawConfig.getAppIDIndex()));
                    userInfoVo.setNum(new Long(1));
                    userInfoVo.setUserTime(userDrawConfig.getProcedureTimeIndex());
                    return userInfoVo;
                })// MDN + appID
                .mapToPair(s -> new Tuple2<>(s.getMobile() + s.getAppId(), s))
                .reduceByKey((v1, v2) -> {
                    v1.setNum(v1.getNum() + v2.getNum());
                    v1.setUserTime(v1.getUserTime() + v2.getUserTime());
                    return v1;
                }).filter(s -> StringUtils.isNotBlank(s._2.getAppId()))
                .collectAsMap()
                .forEach((k, v) -> {
                    //如果不包含 app id 就不统计
                    if (!appMap.containsKey(v.getAppId())) {
                        return;
                    }
                    AppUserWeightVo appUserWeightVo = appMap.get(v.getAppId());
                    // 没有在统计的 app 里 也不统计
                    if (appUserWeightVo == null) {
                        return;
                    }
                    //初次统计，需要初始化信息
                    if (!userDrawMap.containsKey(v.getMobile())) {
                        userDrawMap.put(v.getMobile(), new UserDrawVo(v, appUserWeightVo.getMale(),
                                        appUserWeightVo.getFemale(), appUserWeightVo.getAge1(),
                                        appUserWeightVo.getAge2(), appUserWeightVo.getAge3(),
                                        appUserWeightVo.getAge4(), appUserWeightVo.getAge5()));
                        return;
                    }
                    UserDrawVo userDrawVo = userDrawMap.get(v.getMobile());
                    // 性别权重
                    userDrawVo.protraitSex(appUserWeightVo.getMale(), appUserWeightVo.getFemale(), v.getNum());
                    // 年龄段权重
                    userDrawVo.protraitAge(appUserWeightVo.getAge1(), appUserWeightVo.getAge2(), appUserWeightVo.getAge3(), appUserWeightVo.getAge4(), appUserWeightVo.getAge5(), v.getNum());
                });
        // 遍历打印 结果
        userDrawMap.forEach((k, v) -> System.out.println(k + "   " + v + "    " + Thread.currentThread()));
    }


}
